#pragma once

#include <string>
#include "consumer/client/MessageSubscriber.h"
#include "consumer/client/RouteInfoDownloader.h"
#include "consumer/client/ConsumerMediator.h"

namespace zeroplus::mq
{
    using namespace consumer;
    class ConsumerController
    {
    private:
        std::shared_ptr<consumer::RouteInfoDownloader> routeInfoDownloader_;
        std::shared_ptr<consumer::MessageSubscriber> MessageSubscriber_;
        std::shared_ptr<consumer::ConsumerMediator> mediator_;

    public:
        ConsumerController(const std::string &uri);
        ~ConsumerController();

        void Start(const std::string &topic);

        template <typename MsgT>
        bool Listen(const std::function<void(const MsgT &)> processMsgCb)
        {
            auto handler = [this, processMsgCb](const Bytes &msgBytes)
            {
                Message<Bytes> msg = Message<Bytes>::getDeserializeData(msgBytes);
                if constexpr (std::is_same_v<MsgT, Bytes>)
                {
                    processMsgCb(msgBytes);
                }
                else
                {
                    MsgT message;
                    try
                    {
                        Serializer::convert(msg.getMessage(), message);
                    }
                    catch (const msgpack::v1::type_error &e)
                    {
                        LOG_ERROR("Type error %s", e.what());
                    }
                    catch (const std::exception &e)
                    {
                        LOG_ERROR("Exception: %s", e.what());
                    }
                    processMsgCb(message);
                }
                std::string messageId = msg.getId();
                MessageSubscriber_->SendAckMessage(messageId, transport::ConsumeStatus::CONSUME_SUCCESS); // 模拟消息消费成功
                // LOG_DEBUG("消息已应答: %s", messageId.c_str());
            };
            bool res = MessageSubscriber_->Listen(handler);
            return res;
        }
    };
}
